/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.server;

import java.util.List;
import java.util.concurrent.ExecutorService;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

import org.slf4j.Logger;
import org.sss.common.SentinelThreadPool;
import org.sss.common.io.DBSet;
import org.sss.common.io.IOManager;
import org.sss.common.io.IOThread;
import org.sss.common.io.StreamDBSet;
import org.sss.mapreduce.Cluster;
import org.sss.mapreduce.StorageNode;
import org.sss.util.BlockPolicy;

public class SssServerResource {
  @SuppressWarnings("unused")
  private static Logger logger = SssServer.getLogger(SssServerResource.class);
  private final SentinelThreadPool iotp_r;
  private final SentinelThreadPool iotp_w;
  private final SentinelThreadPool wktp;
  private final ExecutorService tp_combiner;
  private final SssServerConfiguration sssConf;
  private final Cluster cluster;
  private final DBSet       dbForSideData;
  private final StreamDBSet streamDBFroSideData;
  private final IOManager ioManager;

  public SssServerResource(SssServerConfiguration conf, Cluster cluster) throws Exception {
    this.sssConf = conf;
    this.cluster = cluster;
    this.iotp_r = IOThread.createPool(sssConf.ioConf, sssConf.input_threads_max,  
        sssConf.inputtp_workqueue_limit, cluster.getStorageNodes(), "IOR", false);
    this.iotp_w = IOThread.createPool(sssConf.ioConf, sssConf.output_threads_max, sssConf
        .outputtp_workqueue_limit, cluster.getStorageNodes(), "IOW", false);
    this.wktp = WKThread.createPool(sssConf, "WKR");
    this.tp_combiner = new ThreadPoolExecutor(
        sssConf.wk_threads_max, sssConf.wk_threads_max,
        0L, TimeUnit.MILLISECONDS,
        new LinkedBlockingQueue<Runnable>(sssConf.wktp_workqueue_limit),
        new BlockPolicy());
    if (conf.ioConf.stream_protocol) {
      this.dbForSideData = null;
      this.streamDBFroSideData = new StreamDBSet(true);
    }
    else {
      this.dbForSideData = new DBSet(sssConf.ioConf.db_factory, sssConf.ioConf, true);
      this.streamDBFroSideData = null;
    }
    this.ioManager = new IOManager(sssConf.ioConf);
  }

  public SssServerResource(int id) throws Exception {
    this(readConf(id), Cluster.loadDefaultConfigFile());
  }

  public SentinelThreadPool getIOTPRead() {
    return iotp_r;
  }
  public SentinelThreadPool getIOTPWrite() {
    return iotp_w;
  }
  public SentinelThreadPool getWKTP() {
    return wktp;
  }
  public ExecutorService getCombinerTP() {
    return tp_combiner;
  }
  public SssServerConfiguration getServerConfiguration() {
    return sssConf;
  }
  public Cluster getCluster() {
    return cluster;
  }
  public int getStorageNodeCount() {
    return cluster.getStorageNodes().size();
  }
  public List<StorageNode> getStorageNodeList() {
    return cluster.getStorageNodes();
  }
  public DBSet getDBSetForSideData() {
    assert dbForSideData != null;
    dbForSideData.incRef();
    return dbForSideData;
  }
  public StreamDBSet getStreamDBSetForSideData() {
    assert streamDBFroSideData != null;
    streamDBFroSideData.incRef();
    return streamDBFroSideData;
  }
  public IOManager getIOManager() {
    return ioManager;
  }

  public void shutdown() {
    iotp_r.shutdown();
    iotp_w.shutdown();
    wktp.shutdown();
    tp_combiner.shutdown();
  }

  private static SssServerConfiguration readConf(int id) {
    SssServerConfiguration sssConf = SssServerConfiguration.readFile();
    sssConf.setServerId(id);

    return sssConf;
  }
}
